//========================================================================
//Copyright 2007-2010 David Yu dyuproject@gmail.com
//------------------------------------------------------------------------
//Licensed under the Apache License, Version 2.0 (the "License");
//you may not use this file except in compliance with the License.
//You may obtain a copy of the License at 
//http://www.apache.org/licenses/LICENSE-2.0
//Unless required by applicable law or agreed to in writing, software
//distributed under the License is distributed on an "AS IS" BASIS,
//WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//See the License for the specific language governing permissions and
//limitations under the License.
//========================================================================

/*
 * Copyright (c) 2009 Jens Scheffler (appenginefan.com)
 * 
 * Licensed under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in
 * compliance with the License. You may obtain a copy of the
 * License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in
 * writing, software distributed under the License is
 * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
 * CONDITIONS OF ANY KIND, either express or implied. See
 * the License for the specific language governing
 * permissions and limitations under the License.
 */

package com.dyuproject.kvstore;

import java.util.Collections;
import java.util.List;
import java.util.SortedMap;
import java.util.Map.Entry;

import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

/**
 * A simple, map-based store that holds data in memory. Should only be used for
 * unit tests, mostly for two reasons. First of all, this store does not do any
 * defensive copies of its content, so nasty side effects can occur. Second,
 * since all is in memory, nothing is really getting persisted beyond the
 * lifetime of the virtual machine.
 */
public final class MapBasedStore<T> extends AbstractStore<T>
{
    
    static final int MAX_RETRIES = 10;

    final SortedMap<String, byte[]> store = Maps.newTreeMap();

    /**
     * Factory method
     */
    public static <T> MapBasedStore<T> newInstance(Serializer<T> serializer)
    {
        return new MapBasedStore<T>(serializer);
    }

    public MapBasedStore(Serializer<T> serializer)
    {
        super(serializer);
    }

    public synchronized void put(String key, T value)
    {
        Preconditions.checkNotNull(key);
        Preconditions.checkNotNull(value);
        store.put(key, serializer.ser(value));
    }

    public synchronized void delete(String key)
    {
        Preconditions.checkNotNull(key);
        store.remove(key);
    }

    public synchronized void delete(List<String> keys)
    {
        Preconditions.checkNotNull(keys);
        for (String key : keys)
            store.remove(key);
    }
    

    public synchronized T mutate(String key, Function<? super T, ? extends T> mutator)
    {
        Preconditions.checkNotNull(key);
        Preconditions.checkNotNull(mutator);
        T previous = get(key);
        T toSave = mutator.apply(previous);
        if (toSave != null)
        {
            store.put(key, serializer.ser(toSave));
        }
        else if(previous != null)
        {
            store.remove(key);
        }
        return toSave;
    }
    
    public synchronized T mutate(String key, Function<? super T, ? extends T> mutator, 
            int maxRetries)
    {
        return mutate(key, mutator);
    }

    public synchronized T mutateRaw(Mutable<String> mutableKey, 
            Function<? super byte[], ? extends T> mutator)
    {
        Preconditions.checkNotNull(mutableKey);
        Preconditions.checkNotNull(mutator);
        String key = mutableKey.getValue();
        Preconditions.checkNotNull(key);
        
        T toSave = null;
        for(int i=0; i<MAX_RETRIES; i++)
        {
            byte[] previous = store.get(key);
            toSave = mutator.apply(previous);
            String checkKey = mutableKey.getValue();
            // abort
            if(checkKey == null)
                return toSave;
            
            if(checkKey != key)
            {
                key = checkKey;
                continue;
            }
            
            if (toSave != null)
            {
                store.put(key, serializer.ser(toSave));
            }
            else if(previous != null)
            {
                store.remove(key);
            }
            
            break;
        }
        

        return toSave;
    }
    
    public synchronized T mutateRaw(Mutable<String> mutableKey, 
            Function<? super byte[], ? extends T> mutator, int maxRetries)
    {
        return mutateRaw(mutableKey, mutator);
    }

    public synchronized <V> V get(String key, Serializer<V> serializer)
    {
        Preconditions.checkNotNull(key);
        byte[] value = store.get(key);
        return value == null?null:serializer.deser(value);
    }
    
    public synchronized <V> List<V> get(List<String> keys, Serializer<V> serializer)
    {
        Preconditions.checkNotNull(keys);
        if (keys.size() == 0)
            return Collections.emptyList();

        List<V> result = Lists.newArrayListWithCapacity(keys.size());
        for (String k : keys)
        {
            if (k != null)
            {
                byte[] value = store.get(k);
                if (value != null)
                    result.add(serializer.deser(value));
            }
        }
        return result;
    }

    public synchronized <V> List<Entry<String, V>> getEntries(List<String> keys, 
            Serializer<V> serializer)
    {
        Preconditions.checkNotNull(keys);
        if (keys.size() == 0)
            return Collections.emptyList();

        List<Entry<String, V>> result = Lists.newArrayListWithCapacity(keys.size());
        for (String k : keys)
        {
            if (k != null)
            {
                byte[] value = store.get(k);
                if (value != null)
                    result.add(Maps.immutableEntry(k, serializer.deser(value)));
            }
        }
        return result;
    }

    public synchronized <V> List<Entry<String, V>> scanEntries(String start, String end, int max, 
            Serializer<V> serializer)
    {
        SortedMap<String, byte[]> subMap = store.subMap(start, end);
        int capacity = subMap.size() > max?max:subMap.size();
        List<Entry<String, V>> result = Lists.newArrayListWithCapacity(capacity);
        for (Entry<String, byte[]> entry : subMap.entrySet())
        {
            if (result.size() >= max)
            {
                break;
            }
            result.add(Maps.immutableEntry(entry.getKey(), serializer.deser(entry.getValue())));
        }
        return result;
    }

    public synchronized <V> List<Entry<String, V>> scanEntriesReverse(String start, String end, 
            int max, Serializer<V> serializer)
    {
        List<Entry<String, byte[]>> sublist = Lists.newLinkedList(
                store.subMap(start, end).entrySet());
        while (sublist.size() > max)
        {
            sublist.remove(0);
        }
        List<Entry<String, V>> result = Lists.newArrayListWithCapacity(sublist.size());
        for (Entry<String, byte[]> entry : sublist)
        {
            result.add(0, Maps.immutableEntry(entry.getKey(), serializer.deser(entry.getValue())));
        }
        return result;
    }
    
    public synchronized <V> List<V> scan(String start, String end, int max, 
            Serializer<V> serializer)
    {
        SortedMap<String, byte[]> subMap = store.subMap(start, end);
        int capacity = subMap.size() > max?max:subMap.size();
        List<V> result = Lists.newArrayListWithCapacity(capacity);
        for (Entry<String, byte[]> entry : subMap.entrySet())
        {
            if (result.size() >= max)
            {
                break;
            }
            result.add(serializer.deser(entry.getValue()));
        }
        return result;
    }
    
    public synchronized <V> List<V> scanReverse(String start, String end, int max, 
            Serializer<V> serializer)
    {
        List<Entry<String, byte[]>> sublist = Lists.newLinkedList(
                store.subMap(start, end).entrySet());
        while (sublist.size() > max)
        {
            sublist.remove(0);
        }
        List<V> result = Lists.newArrayListWithCapacity(sublist.size());
        for (Entry<String, byte[]> entry : sublist)
        {
            result.add(0, serializer.deser(entry.getValue()));
        }
        return result;
    }

    public synchronized List<String> keyScan(String start, String end, int max)
    {
        SortedMap<String, byte[]> subMap = store.subMap(start, end);
        int capacity = subMap.size() > max?max:subMap.size();
        List<String> result = Lists.newArrayListWithCapacity(capacity);
        for (Entry<String, byte[]> entry : subMap.entrySet())
        {
            if (result.size() >= max)
            {
                break;
            }
            result.add(entry.getKey());
        }
        return result;
    }

    public synchronized List<String> keyScanReverse(String start, String end, int max)
    {
        List<Entry<String, byte[]>> sublist = Lists.newLinkedList(
                store.subMap(start, end).entrySet());

        while (sublist.size() > max)
        {
            sublist.remove(0);
        }
        List<String> result = Lists.newArrayListWithCapacity(sublist.size());
        for (Entry<String, byte[]> entry : sublist)
        {
            result.add(0, entry.getKey());
        }
        return result;
    }
}
